package se.kth.ict.id2203.riwcm;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.beb.BebBroadcast;
import se.kth.ict.id2203.beb.BestEffortBroadcast;
import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.launch.Topology;

public class RIWCM extends ComponentDefinition {
	
	Negative<AtomicRegister> riwcm = provides(AtomicRegister.class);
	Positive<BestEffortBroadcast> beb = requires(BestEffortBroadcast.class);
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	
	private Address self;
	private Topology topology;
	
	// number of registers
	private static final int rnum = 10;
	
	private Map<Integer, Set<Address>> writeSet;
	private Map<Integer, Set<Triple>> readSet;
	private boolean reading[];
	private long reqid[];
	private long v[];
	private long ts[];
	private long mrank[];
	private long writeval[];
	private long readval[];
	private long i;
	
	private static final Logger logger =
			LoggerFactory.getLogger(RIWCM.class);
	
	public RIWCM() {
		logger.info("RIWC created.");
		subscribe(handleInit, control);
		subscribe(handleReadReq, riwcm);
		subscribe(handleWriteReq, riwcm);
		subscribe(handleAck, pp2p);
		subscribe(handleWriteMsg, beb);
		subscribe(handleReadMsg, beb);
		subscribe(handleReadVal, pp2p);
		
	}
	
	private long rank(Address address) {
		return address.getId();
	}
	
	private Triple highest(Set<Triple> tpset) {
		
		Triple high = new Triple(0, 0, 0);
		
		for (Triple tp : tpset) {
			if (tp.getT() > high.getT()) {
				high = tp;
			} else if (tp.getT() == high.getT()) {
				if (tp.getRk() > high.getRk()) {
					high = tp;
				}
			}
		}
		
		return high;
	}
	

	Handler<RIWCMInit> handleInit = new Handler<RIWCMInit>() {
		public void handle(RIWCMInit event) {
			
			logger.info("RIWCM started.");
			topology = event.getTopology();
			self = topology.getSelfAddress();
			
			writeSet = new HashMap<Integer, Set<Address>>();
			readSet = new HashMap<Integer, Set<Triple>>();
			reading = new boolean[rnum];
			reqid = new long[rnum];
			v = new long[rnum];
			ts = new long[rnum];
			mrank = new long[rnum];
			readval = new long[rnum];
			writeval = new long[rnum];
					
			i = rank(self);
			for (int r = 0; r < rnum; r++) {
				
				writeSet.put(r, new HashSet<Address>());
				readSet.put(r, new HashSet<Triple>());
				reading[r] = false;
				reqid[r] = 0;
				v[r] = 0;
				ts[r] = 0;
				mrank[r] = 0;
				readval[r] = 0;
				writeval[r] = 0;
			}
		}
	};
	

	Handler<ReadRequest> handleReadReq = new Handler<ReadRequest>() {
		public void handle(ReadRequest event) {
			
			int r = event.getRegister();
			logger.info("ReadRequest(" + r + ") received!");
			
			reqid[r] = reqid[r] + 1;
			reading[r] = true;
			readSet.get(r).clear();
			writeSet.get(r).clear();

			// trigger BebBroadcast
			trigger(new BebBroadcast(new ReadMessage(self, r, reqid[r])), beb);
		}
	};
	
	Handler<WriteRequest> handleWriteReq = new Handler<WriteRequest>() {
		public void handle(WriteRequest event) {
			
			int r = event.getRegister();
			long val = event.getVal();
			logger.info("WriteRequest(" + r + ", " + val + ") received!");
			
			reqid[r] = reqid[r] + 1;
			writeval[r] = val;
			readSet.get(r).clear();
			writeSet.get(r).clear();
			// trigger BebBroadcast
			trigger(new BebBroadcast(new ReadMessage(self, r, reqid[r])), beb);
			
		}
	};
	
	Handler<ReadMessage> handleReadMsg = new Handler<ReadMessage>() {
		public void handle(ReadMessage event) {
			
			Address source = event.getSource();
			int r = event.getRegister();
			long id = event.getId();
			logger.info("ReadMessage(" + r + ", " + id + ") received!");
			logger.info("Sending ReadValueMessage to " + source);
			
			trigger(new Pp2pSend(source, 
					new ReadValueMessage(self, r, id, ts[r], mrank[r], v[r])), 
					pp2p);
		}
	};
	
	Handler<ReadValueMessage> handleReadVal = new Handler<ReadValueMessage>() {
		public void handle(ReadValueMessage event) {
			
			Address source = event.getSource();
			int r = event.getRegister();
			long id = event.getId();
			long t = event.getT();
			long j = event.getJ();
			long val = event.getVal();
			
			logger.info("ReadValueMessage(" + r + ", " + id + ") received from " + source + "!");
			
			if (id == reqid[r]) {
				readSet.get(r).add(new Triple(t, j, val));
				
				logger.info("readSet size: " + readSet.get(r).size());
			}
			
		//	for (int rr = 0; rr < rnum; rr++) {
				if (readSet.get(r).size() > topology.getNodeCount() / 2) {
					
					Triple temp = highest(readSet.get(r));
					long tt = temp.getT();
					long rk = temp.getRk();
					long vv = temp.getVal();
					
					readval[r] = vv;
					
	
					if (reading[r] == true) {
						trigger(new BebBroadcast(new WriteMessage(
								self, r, reqid[r], tt, rk, readval[r])), beb);
					} else {
						trigger(new BebBroadcast(new WriteMessage(
								self, r, reqid[r], tt+1, i, writeval[r])), beb);
					}
					
					// what I add to the algorithm
					readSet.get(r).clear();
				}
		//	}
			
		}
		
	};
	
	Handler<WriteMessage> handleWriteMsg = new Handler<WriteMessage>() {
		public void handle(WriteMessage event) {
			
			Address source = event.getSource();
			int r = event.getRegister();
			long id = event.getId();
			long t = event.getT();
			long j = event.getJ();
			long val = event.getVal();
						
			logger.info("WriteMessage(" + r + ", " + id + ", " + 
					t + ", " + j + ", " + val + ")Received!");
			
			if (t > ts[r]) {
				v[r] = val;
				ts[r] = t;
				mrank[r] = j;
			} else if (t == ts[r]) {
				if (j > mrank[r]) {
					v[r] = val;
					ts[r] = t;
					mrank[r] = j;
				}
			}
			
			trigger(new Pp2pSend(source, new AckMessage(self, r, id)), pp2p);
			
		}
	};
	
	Handler<AckMessage> handleAck = new Handler<AckMessage>() {
		public void handle(AckMessage event) {
			
			Address source = event.getSource();
			int r = event.getRegister();
			long id = event.getId();
			
			logger.info("AckMessage received from " + source);
			
			if (id == reqid[r]) {
				writeSet.get(r).add(source);
				
				logger.info("writeSet size: " + writeSet.get(r).size());
			}

			// upon exists r such that

			if (writeSet.get(r).size() > topology.getNodeCount() / 2) {
				
				if (reading[r] == true) {
					reading[r] = false;
					trigger(new ReadResponse(r, readval[r]), riwcm);
				} else {
					trigger(new WriteResponse(r), riwcm);
				}
				
				// what I add to the algorithm
				writeSet.get(r).clear();
			}

		}
	};
	
	
}
